#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>
#include <thrift/concurrency/ThreadManager.h>
#include <thrift/concurrency/PosixThreadFactory.h>

#include <boost/thread/mutex.hpp>
#include "conn-pool/ConnPool.h"
#include "boost/bind.hpp"
#include "Serv.h"

#include "ServerManager.h"
#include "trace_worker.h"
#include "JsonHelper.h"
#include "ClientManager.h"
#include "IweChenCfg.h"

using namespace ::apache::thrift;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::server;
using namespace ::apache::thrift::concurrency;

static boost::mutex g_insMutexCalc;

class ServHandler : virtual public ServIf {
 public:
    ServHandler()
    {
    }

    virtual void thriftMethod(const std::string& openId, const std::string& method, const std::string& reqArgs)
    {   trace_worker();
        trace_printf("openId.c_str(), method.c_str(), reqArgs.c_str()  %s  %s  %s", openId.c_str(), method.c_str(), reqArgs.c_str());
        rapidjson::Document document;
        if (!document.Parse(reqArgs.c_str()).HasParseError())
        {
            if (openId.size() > 0)
            {
                document.HasMember("openId") ? document["openId"].SetString(openId.c_str(), document.GetAllocator()) : document.AddMember("openId", rapidjson::Value(openId.c_str(), document.GetAllocator()), document.GetAllocator());
            }
            rapidjson::Value returnValue(rapidjson::kObjectType);
            if (CServerManager::instance()->thriftMethod(method, document, returnValue) == true && !returnValue.IsNull())
            {   trace_printf("NULL");
                CClientManager::instance()->SendWSData(document, returnValue);
            }
        }
    }

    virtual void thriftMethodSyn(std::string& _return, const std::string& openId, const std::string& method, const std::string& reqArgs)
    {   trace_worker();
        trace_printf("openId.c_str(), method.c_str(), reqArgs.c_str()  |%s|  |%s|  |%s|", openId.c_str(), method.c_str(), reqArgs.c_str());
        rapidjson::Document document;
        if (!document.Parse(reqArgs.c_str()).HasParseError())
        {
            if (openId.size() > 0)
            {
                document.HasMember("openId") ? document["openId"].SetString(openId.c_str(), document.GetAllocator()) : document.AddMember("openId", rapidjson::Value(openId.c_str(), document.GetAllocator()), document.GetAllocator());
            }
            int serialNumber = document.HasMember("serialNumber") ? document["serialNumber"].GetInt() : -1;

            rapidjson::Value returnValue(rapidjson::kObjectType);
            if (CServerManager::instance()->thriftMethod(method, document, returnValue) == true && !returnValue.IsNull())
            {   trace_printf("NULL");
                if (serialNumber > 0)
                {
                    returnValue.AddMember("serialNumber", serialNumber, document.GetAllocator());
                }
                rapidjson::StringBuffer buffer;
                _return = CJsonHelper::toString(returnValue, buffer);
            }
        }
    }
};

CServerManager* CServerManager::_instance = NULL;

CServerManager* CServerManager::instance() 
{	
	if (NULL == _instance)
	{
		boost::unique_lock<boost::mutex> guardMutex(g_insMutexCalc);
		if (NULL == _instance)
		{
			_instance = new CServerManager;
		}
	}
	return _instance;
}


CServerManager::CServerManager()
{
}

CServerManager::~CServerManager()
{
}

bool CServerManager::RegisterMethod(const char *methodName, Method method)
{
    m_methodMap[methodName] = method;
    return true;
}

bool CServerManager::thriftMethod(const std::string& method, rapidjson::Document &document, rapidjson::Value &returnValue)
{
    Method &methodfunction = m_methodMap[method];
    if (methodfunction)
    {
        methodfunction(document, returnValue);
        return true;
    }
    return false;
}

void CServerManager::ConnPoolInit(const rapidjson::Value &cfgValue)
{
    CRedisConnPool::instance()->Init(cfgValue["REDIS"]);
    CSqlConnPool::instance()->Init(cfgValue["MYSQL"]);
}

bool CServerManager::StartServer()
{   trace_worker();

    const rapidjson::Value &cfgValue = CWeChenCfg::instance()->Cfg();

    const rapidjson::Value &threadNumsValue = cfgValue["SERV"]["thread_nums"];
    const rapidjson::Value &portsValue = cfgValue["SERV"]["ports"];
    if (threadNumsValue.Size() != portsValue.Size())
    {
        printf("threadNumsValue.Size() != portsValue.Size()  %d != %d", threadNumsValue.Size(), portsValue.Size());
        return false;
    }
    int forSize = threadNumsValue.Size();
    
    boost::shared_ptr<ServerHandle> serverHandle;
    for (int i=0; i<forSize; ++i)
    {        
        int thread_num = threadNumsValue[i].GetInt();
        int port = portsValue[i].GetInt();
        boost::shared_ptr<ServHandler> handler(new ServHandler());
        boost::shared_ptr<TProcessor> processor(new ServProcessor(handler));
        boost::shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());  
        boost::shared_ptr<ThreadManager> threadManager = ThreadManager::newSimpleThreadManager(thread_num);  
        boost::shared_ptr<PosixThreadFactory> threadFactory = boost::shared_ptr<PosixThreadFactory > (new PosixThreadFactory());  
        threadManager->threadFactory(threadFactory);  
        threadManager->start();
        
        serverHandle = boost::shared_ptr<ServerHandle>(new ServerHandle());
        serverHandle->nonblockingServer = boost::shared_ptr<TNonblockingServer>(new TNonblockingServer(processor, protocolFactory, port, threadManager));
        serverHandle->serverThread =  boost::shared_ptr<boost::thread>(new boost::thread(boost::bind(&TNonblockingServer::serve, &(*serverHandle->nonblockingServer))));
        m_serverHandleVector.push_back(serverHandle);
    }
    ConnPoolInit(cfgValue);
    return true;
}




